package org.zjt.spark.book

import org.apache.spark.mllib.classification.LogisticRegressionWithLBFGS
import org.apache.spark.mllib.feature.{HashingTF, IDF}
import org.apache.spark.mllib.regression.LabeledPoint
import org.apache.spark.{SparkConf, SparkContext}

/**
  * 邮件垃圾分类    tf-ide + 特征逻辑回归
  *
  *   1、中文分词器先分词，将分词后的数据归档到文件中（空格隔开）。
  *   2、利用TF将词频转为向量
  *   3、对该向量几何求出逻辑回归模型。
  *   4、利用模型预期结果。是否为垃圾邮件。
  *
  */
object EmailClassification extends App{
  var sparkConf = new SparkConf().setMaster("local[2]").setAppName("EmailClassification")
  val sc = new SparkContext(sparkConf)

  // tf转化器
  val tf:HashingTF = new HashingTF(1000)
  val df = new IDF()

  //中文必须调用中文分词器来分词。

  //将邮件中的词语分词，并计算词频向量特征向量。
  val normal = sc.textFile("D:\\Idea workspace\\scala-demo\\src\\main\\resource\\mllib\\nomal.txt").map{
    line => tf.transform(line.split(" "))
  }

  val positive = df.fit(normal).transform(normal).map{
    line => LabeledPoint(1,line)
  }

  val spam = sc.textFile("D:\\Idea workspace\\scala-demo\\src\\main\\resource\\mllib\\spam.txt").map{
    line => tf.transform(line.split(" "))
  }

  val negative = df.fit(spam).transform(spam).map{
    line => LabeledPoint(0,line)
  }

  //合并两个特征集
  val training = negative.union(positive).cache()

  val model = new LogisticRegressionWithLBFGS().setNumClasses(2).run(training)

  println(if( model.predict(tf.transform("Viagra real cheap!  Send money right".split(" "))) == 0) "垃圾邮件" else "正常邮件" )
  println(if( model.predict(tf.transform("user list, This is my first question to".split(" "))) == 0) "垃圾邮件" else "正常邮件" )
}
